package work_pattern.publisher_confirms;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import work_pattern.Constant;

import java.io.IOException;
import java.util.Collections;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.TimeoutException;

public class Producer {
    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        Channel channel = Constant.createChannel();
        // 设置信道为 confirm 模式
        channel.confirmSelect();
        channel.queueDeclare(Constant.PUBLISH_CONFIRM_QUEUE, true, false, true, null);
//        channel.basicPublish("", Constant.PUBLISH_CONFIRM_QUEUE, null, "message...".getBytes());
//        /**
//         * 等待确认消息
//         * 未确认的消息存放在 SortedSet<Long> unconfirmedSet =
//         *             Collections.synchronizedSortedSet(new TreeSet<Long>()) 集合中
//         */
//        channel.waitForConfirms(5000);
//        System.out.println("消息确认");
        SortedSet<Long> confirmSet = Collections.synchronizedSortedSet(new TreeSet<>());
        channel.addConfirmListener(new ConfirmListener() {
            @Override
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                System.out.println("deliveryTag: " + deliveryTag + " multiple: " + multiple);
                if (multiple) {
                    confirmSet.headSet(deliveryTag + 1).clear();
                } else {
                    confirmSet.remove(deliveryTag);
                }
            }

            @Override
            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                System.out.println("消息 " + deliveryTag + " 处理失败...");
                if (multiple) {
                    confirmSet.headSet(deliveryTag + 1).clear();
                } else {
                    confirmSet.remove(deliveryTag);
                }
            }
        });
        for (int i = 0; i < 20; i++) {
            String msg = "message " + i;
            // 获取下次发送消息序列化
            long nextPublishSeqNo = channel.getNextPublishSeqNo();
            channel.basicPublish("", Constant.PUBLISH_CONFIRM_QUEUE, null, msg.getBytes());
            confirmSet.add(nextPublishSeqNo);
        }
    }
}
